/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
/**
 * Software License, Version 1.0
 * 
 * Copyright 2004 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.gridapps.nbgridftp.subscriber;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.Properties;

import javax.jms.BytesMessage;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;

import cgl.narada.event.NBEvent;
import cgl.narada.event.TemplateInfo;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.jms.NBJmsInitializer;
import cgl.narada.matching.Profile;
import cgl.narada.service.ServiceBulletin;
import cgl.narada.service.ServiceException;
import cgl.narada.service.ServiceNotification;
import cgl.narada.service.Task;
import cgl.narada.service.client.ClientService;
import cgl.narada.service.client.EventConsumer;
import cgl.narada.service.client.EventProducer;
import cgl.narada.service.client.NBEventListener;
import cgl.narada.service.client.NBRecoveryListener;
import cgl.narada.service.client.NBRecoveryNotification;
import cgl.narada.service.client.SessionService;
import cgl.narada.service.qos.ConsumerConstraints;
import cgl.narada.service.qos.ProducerConstraints;
/**
   @author Sang Boem Lim
   $Date$
   $Revision$
 */

class NB2ServerControl extends Thread implements javax.jms.MessageListener,
NBEventListener, NBRecoveryListener, ServiceBulletin{

    private int port;
    private String host;
    private String hostPort;

    private static InputStream fromServer;
    private static OutputStream toServer;

    private Properties connectionProperties;

    private final String FSC     = "From server control: ";
    private final String TEMPDIR = "/tmp/NB/nb2s/";

    //naradabrokering..
    private ClientService clientService;
    private String moduleName= "CoalescingClient: ";
    private EventConsumer consumer;
    private EventProducer producer;
    private ProducerConstraints producerConstraints;

    public NB2ServerControl(Properties connectionProperties, 
                            String host, int port) {

        this.host     = host;
        this.port     = port;

        this.connectionProperties = connectionProperties;
    }

    public void run() {

        startService();
        getFromServer();
    }

    private void startService(){

        // NB set up..
        try {
            String transportType = "niotcp";
            int entityId = 7777;
            int templateId = 12345;

            initializeChatSession((TopicConnectionFactory) 
                                  new NBJmsInitializer(connectionProperties, 
                                                        transportType, entityId).lookup());
            clientService = SessionService.getClientService(entityId, this);
            clientService.initializeBrokerCommunications(connectionProperties,
                                                         transportType);
            initializeProducerAndConsumer(templateId);
        } catch (Exception e){
            System.err.println(e);
            System.exit(1);
        }
    }

    private static NB2ServerData data;
    public void getFromServer() {

        while (true) {
            isRecovered = false;
            isFirstTime = true;
            isSent = false;
            try {
                // Connect to the GridFTP server.
                Socket s = new Socket(host,port); 
                fromServer = s.getInputStream();
                toServer   = s.getOutputStream();
                
                clientName = s.getInetAddress().getHostAddress() + ":" + 
                    s.getPort();

                int bytes_read;
                byte[] request = new byte[1024];
                
                bytes_read = fromServer.read(request);

                data = new NB2ServerData();
                data.start();

                hostPort = data.getHostPort();

                writeControlMessage(new String(request, 0, bytes_read) + 
                                    ":::" + hostPort);

                while((bytes_read = fromServer.read(request)) != -1)
                    writeMessage(request, bytes_read, FSC);

                if (s != null) {
                    s.close();
                    s = null;
                }

                cleanup();
            } catch (Exception e){
                cleanup();
            }
        }
    }

    public static void cleanup() {

        try {
            if (toServer != null) {
                toServer.close();
                toServer = null;
            }
            if (fromServer != null) {
                fromServer.close();
                fromServer = null;
            }

        } catch (IOException e) {}

        if (data != null) {
            data.close();
            data = null;
        }
        
        isNotSet = true;
    }

    private static TopicSession pubSession;
    private static TopicSession subSession;
    private static TopicPublisher publisher;
    private static TopicConnection connection;
    private static String clientName;

    /* Create and send message using topic publisher */
    public void writeControlMessage(String str)
        throws JMSException {
        
        TextMessage message = pubSession.createTextMessage();
        message.setText(str);
        publisher.publish(message);
    }

    /* Create and send message using topic publisher */
    public static void writeMessage(byte[] buf, int length, String msg)
        throws JMSException {

        byte[] header = (msg + clientName).getBytes();

        BytesMessage byteMessage = pubSession.createBytesMessage();
        byteMessage.writeInt(header.length);
        byteMessage.writeBytes(header);
        byteMessage.writeInt(length);
        byteMessage.writeBytes(buf);
        
        publisher.publish(byteMessage); 
    }

    private static boolean isNotSet = true;
    /* Receive message from topic subscriber */
    public void onMessage(Message message) {

        try {
            if (message.getJMSType().equals("BytesMessage")) {

                BytesMessage byteMessage = (BytesMessage) message;

                byte[] byteHeader = new byte[byteMessage.readInt()];
                byteMessage.readBytes(byteHeader);

                String header = new String(byteHeader);

                int length = byteMessage.readInt();

                if(header.startsWith("CONTROL:")) {

                    byte[] reply = new byte[length];
                    byteMessage.readBytes(reply);

                    try{
                        while(toServer == null) Thread.sleep(1);
                    } catch (InterruptedException err){}
                    
                    toServer.write(reply, 0, length);
                    toServer.flush();
                }  else if(header.startsWith("AUTH:")) {

                    byte[] reply = new byte[length];
                    byteMessage.readBytes(reply);
                    
                    NB2ServerData.write2Server(reply, length);

                } else if (header.startsWith("NORMAL:")) {
                    if (toServer != null) {
                        toServer.close();
                        toServer = null;
                    }
                    if (fromServer != null) {
                        fromServer.close();
                        fromServer = null;
                    }

                } else if (header.startsWith("ABNORMAL:")) {
                    
                    if (isNotSet) {
                        if (toServer != null) {
                            toServer.close();
                            toServer = null;
                        }
                        if (fromServer != null) {
                            fromServer.close();
                            fromServer = null;
                        }
                        
                        writeMessage(new byte[0], 0, "SERVER CLOSED: ");
                    }
                } else if(header.startsWith("INTER:")) {
                    
                    if (toServer != null) {
                        toServer.close();
                        toServer = null;
                    }
                    if (fromServer != null) {
                        fromServer.close();
                        fromServer = null;
                    }

                    if((data != null) && (!data.isFinished())) {
                        try{
                            Thread.sleep(1000);
                        } catch (InterruptedException err){}
                    }
                } else if(header.startsWith("WELCOME:")) {

                    writeControlMessage(":::" + hostPort);

                    byte[] reply = new byte[length];
                    byteMessage.readBytes(reply);

                    try{
                        while(toServer == null) Thread.sleep(1);
                    } catch (InterruptedException err){}

                    toServer.write(reply, 0, length);
                    toServer.flush();
                } 
            } 
        } catch(JMSException jmse){ 
            jmse.printStackTrace(); 
        } catch(IOException e) {
            e.printStackTrace();
        }
    }

    private void startConnection() throws JMSException {
        connection.start();
    }
  
    private void stopConnection() throws JMSException {
        connection.stop();
    }

    /* Close the JMS connection */
    private void close() throws JMSException {
        connection.close();
    }

    /* Initializes the instance variables */
    private void set(TopicConnection con, TopicSession pubSess,
                     TopicSession subSess, TopicPublisher pub) {

        this.connection = con;
        this.pubSession = pubSess;
        this.subSession = subSess;
        this.publisher = pub;
    }
  
    private void initializeChatSession(TopicConnectionFactory conFactory) 
        throws Exception {

        // Create a JMS connection
        TopicConnection connection = 
            conFactory.createTopicConnection("guest","password");
    
        // Create a JMS session object
        TopicSession pubSession =
            connection.createTopicSession(false, javax.jms.Session.AUTO_ACKNOWLEDGE);

        TopicSession subSession =
            connection.createTopicSession(false, javax.jms.Session.AUTO_ACKNOWLEDGE);
    
        String serverString = "/slim/GridFTP/server";
        String clientString = "/slim/GridFTP/client";

        // Lookup a JMS topic
        Topic serverTopic = pubSession.createTopic(serverString);  
        Topic clientTopic = pubSession.createTopic(clientString);  
    
        // Create a JMS publisher and subscriber
        TopicPublisher publisher = pubSession.createPublisher(serverTopic);
        TopicSubscriber subscriber = subSession.createSubscriber(clientTopic);
    
        // Set a JMS message listener
        subscriber.setMessageListener(this);
    
        // Intialize the Chat application
        set(connection, pubSession, subSession, publisher);
    
        // Start the JMS connection; allows messages to be delivered
        connection.start();
    }

    public void initializeProducerAndConsumer(int templateId) 
        throws ServiceException {
        Profile profile = 
            clientService.createProfile(TemplateProfileAndSynopsisTypes.STRING, 
                                        "Movies/Casablanca");
        consumer = clientService.createEventConsumer(this);
    
        ConsumerConstraints constraints = 
            consumer.createConsumerConstraints(profile);
        constraints.setReceiveReliably(templateId);
        constraints.setReceiveAfterCoalescingFragments();
        consumer.subscribeTo(profile, constraints);

        long recoveryId = consumer.recover(templateId, this);
        producer = clientService.createEventProducer();
        producer.setSuppressRedistributionToSource(true);
        TemplateInfo templateInfo = 
            clientService.createTemplateInfo(12345, 
                                             TemplateProfileAndSynopsisTypes.STRING,
                                             "Movies/Casablanca");
        producerConstraints = 
            producer.createProducerConstraints(templateInfo);
    }
  

    public void publishEvent(String stringMsg) throws ServiceException {
        if (stringMsg.equals("")) {
            stringMsg += System.currentTimeMillis();
        }
        producer.generateEventIdentifier(true);
        producer.setTemplateId(12345);
        producer.setDisableTimestamp(false);
        NBEvent nbEvent = 
            producer.generateEvent(TemplateProfileAndSynopsisTypes.STRING,
                                   "Movies/Casablanca",
                                   stringMsg.getBytes());
        producer.publishEvent(nbEvent, producerConstraints);
    }
    
    private boolean isSent = false;
    public void onEvent(NBEvent nbEvent) {
        System.out.println("\n\n\n" + moduleName + "Received COALESCED file [" + 
                           new String( nbEvent.getContentPayload() )   +
                           "] of length=" + 
                           nbEvent.getEventHeaders().getOriginalPayloadLength());

        if (progress != null) progress.setDone();
        isSent = true;
        NB2ServerData.sendData(new String(nbEvent.getContentPayload()), 
                               host, port);
    }
        
  
    boolean isRecovered = false;
    /** Upon completion of the attempt to recover, this method is invoked on
     the listener that was registered with the */
    public void onRecovery(NBRecoveryNotification recoveryNotification) {
        isRecovered = true;
        System.out.println("\n\n\n\n" + moduleName + recoveryNotification +"\n\n");
    }


    public void onServiceException(ServiceException serviceException) {
        //System.out.println(moduleName + serviceException);
    }

    private boolean isFirstTime = true;
    private NBProgress progress;
    public void onServiceNotification(ServiceNotification serviceNotification) {
        if (serviceNotification.hasTaskInfo()) {
            Task task = serviceNotification.getTaskInfo();
            
            boolean isSubtasks = task.hasSubTasks();
            
            int subTasks = 1;
            if(isSubtasks)
                subTasks = task.getNumOfSubTasks();
            
            if(task.getStatus() == Task.COMPLETED) {
                if (progress != null) progress.setDone();
                isSent = true;
            } else if(!isSent) {
                if(isFirstTime) {
                    isFirstTime = false;
                    progress = new NBProgress(subTasks);
                    
                    if(isSubtasks) {
                        for(int i = 0; i < subTasks; i++)
                            progress.setup(i, task.getSubTask(i).getNumOfWorkUnits());
                    } else 
                        progress.setup(0, task.getNumOfWorkUnits());
                    
                    progress.start();
                }
                
                if(isSubtasks) {
                    for(int i = 0; i < subTasks; i++)
                        progress.update(i, task.getSubTask(i).getNumOfWorkUnitsCompleted());
                } else 
                    progress.setup(0, task.getNumOfWorkUnitsCompleted());
            }
            //System.out.println(task.toString());
        }
    }
    //********* NB end *************************************8
}





